package com.freez.spark.core.demo;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import scala.Tuple3;

import java.util.ArrayList;
import java.util.List;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: AnalysisShop.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:17
 * Description: 购物栏分析
 */
public class AnalysisShop {
    public static void main(String[] args) {
        String appName = "SparkDemo-AnalysisShop";
        String master = "local";
        String path = "./src/main/resources/data/core/demo/sparkMBA.txt";
        analysisShop(appName, master, path);
    }

    public static void analysisShop(String appName, String master, String path) {
        SparkConf conf = new SparkConf();
        conf.setAppName(appName).setMaster(master);
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> javaRDD = jsc.textFile(path);
        //生成频繁模式 [a,b,c]=> [a]:1,[b]:1,[c]:1,[a,b]:1,[a,c]:1,[b,c]:1,[a,b,c]:1
        JavaPairRDD<List<String>, Integer> patterns = javaRDD.flatMapToPair(s -> {
            List<String> list = toList(s);
            List<List<String>> combinations = getSubsets(list);
            List<Tuple2<List<String>, Integer>> result = new ArrayList<>();
            for (List<String> combList : combinations) {
                if (combList.size() > 0) {
                    result.add(new Tuple2<>(combList, 1));
                }
            }
            return result.iterator();
        });
        //reduceByKey归约
        JavaPairRDD<List<String>, Integer> combined = patterns.reduceByKey((i1, i2) -> i1 + i2);
        //生成所有子模式，如[a,b,c]:2 => [a,b,c]:(null,2), [a,b]:([a,b,c],2), [a,c]:([a,b,c],2), [b,c]:([a,b,c],2)
        JavaPairRDD<List<String>, Tuple2<List<String>, Integer>> subPatterns = combined.flatMapToPair(pattern -> {
            List<Tuple2<List<String>, Tuple2<List<String>, Integer>>> result = new ArrayList<>();
            List<String> list = pattern._1;
            Integer frequency = pattern._2;
            result.add(new Tuple2<>(list, new Tuple2<>(null, frequency)));
            if (1 == list.size()) {
                return result.iterator();
            }
            for (int i = 0; i < list.size(); i++) {
                List<String> subList = removeOneItem(list, i);
                result.add(new Tuple2<>(subList, new Tuple2<>(list, frequency)));
            }
            return result.iterator();
        });
        //组合所有子模式 ([a,b],[(null, 2), ([a,b,d], 1), ([a,b,c], 1)])
        JavaPairRDD<List<String>, Iterable<Tuple2<List<String>, Integer>>> rules = subPatterns.groupByKey();
        //生成关联规则
        JavaRDD<List<Tuple3<List<String>, List<String>, Double>>> assocRules = rules.map(rule -> {
            List<Tuple3<List<String>, List<String>, Double>> result = new ArrayList<>();
            List<String> fromList = rule._1;
            Iterable<Tuple2<List<String>, Integer>> to = rule._2;
            List<Tuple2<List<String>, Integer>> toList = new ArrayList<>();
            Tuple2<List<String>, Integer> fromCount = null;
            for (Tuple2<List<String>, Integer> t2 : to) {
                if (null == t2._1) {
                    fromCount = t2;
                } else {
                    toList.add(t2);
                }
            }
            if (toList.isEmpty()) {
                return result;
            }
            //计算置信度 Y的支持度/X的支持度
            for (Tuple2<List<String>, Integer> t2 : toList) {
                double confidence = t2._2 / (double) fromCount._2;
                List<String> t2List = new ArrayList<>(t2._1);
                t2List.removeAll(fromList);
                result.add(new Tuple3<>(fromList, t2List, confidence)); // X -> Y : confidence
            }
            return result;
        });
        //打印结果
        List<List<Tuple3<List<String>, List<String>, Double>>> result = assocRules.collect();
        for (List<Tuple3<List<String>, List<String>, Double>> r1 : result) {
            System.out.println(r1);
        }
    }

    public static List<String> toList(String line) {
        String[] items = line.trim().split(",");
        List<String> list = new ArrayList<>();
        for (String item : items) {
            list.add(item);
        }
        return list;
    }

    public static List<String> removeOneItem(List<String> list, int i) {
        if ((null == list) || (list.isEmpty())) {
            return list;
        }
        if ((i < 0) || (i > (list.size() - 1))) {
            return list;
        }
        List<String> clonedList = new ArrayList<>(list);
        clonedList.remove(i);
        return clonedList;
    }

    /**
     * 获得集合所有子集
     */
    public static List<List<String>> getSubsets(List<String> elements) {
        List<List<String>> allSubsets = new ArrayList<>();
        int max = 1 << elements.size();
        for (int i = 0; i < max; i++) {
            int index = 0;
            int k = i;
            List<String> list = new ArrayList<>();
            while (k > 0) {
                if ((k & 1) > 0) {
                    list.add(elements.get(index));
                }
                k >>= 1;
                index++;
            }
            allSubsets.add(list);
        }
        return allSubsets;
    }
}